package com.hadoop.mr.pagerank;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;


/**
 * <p>pageRank计算的Reduce</p>
 *
 * @author Andy
 * @date 2017/10/26
 */
public class PageRankReduce extends Reducer<Text, Text, Text, Text> {
    public static final double CONVERGENCE_SCALING_FACTOR = 1000.0;
    public static final double DAMPING_FACTOR = 0.85;
    public static String CONF_NUM_NODES_GRAPH = "pagerank.numnodes";
    private int numberOfNodesInGraph;

    public enum Counter {
        CONV_DELTAS
    }

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        numberOfNodesInGraph = context.getConfiguration().getInt(CONF_NUM_NODES_GRAPH, 0);
    }

    private Text outVal = new Text();

    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        System.out.println("input -> K[" + key + "]");
        //∑pg
        double summedPageRanks = 0;
        Node originNode = new Node();
        for (Text value : values) {
            System.out.println("  input -> V[" + value + "]");

            Node node = Node.fromMr(value.toString());
            if (node.getAdjacentNodeNames() != null) {
                originNode = node;
            } else {
                summedPageRanks += node.getPageRank();
            }
        }

        double dampingFactor = (1.0 - DAMPING_FACTOR) / numberOfNodesInGraph;

        double newPageRank = dampingFactor + DAMPING_FACTOR * summedPageRanks;

        double delta = originNode.getPageRank() - newPageRank;
        originNode.setPageRank(newPageRank);

        outVal.set(originNode.toString());
        System.out.println("  output -> K[" + key + "],V[" + outVal + "]");

        context.write(key, outVal);

        int scaledDelta = Math.abs((int) (delta * CONVERGENCE_SCALING_FACTOR));
        System.out.println("Delta = " + scaledDelta);
        context.getCounter(Counter.CONV_DELTAS).increment(scaledDelta);
    }
}
